package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo27RDDPartitions {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf() // 对Spark进行配置
    conf.setAppName("Demo27RDDPartitions") // 设置程序名称
    conf.setMaster("local") // 设置Spark的运行方式，如果是本地 则设为local即可

    // 设置spark默认并行度
    conf.set("spark.default.parallelism", "4")

    val sc: SparkContext = new SparkContext(conf)

    /**
     * 第一个RDD一般是读取HDFS生成
     * 它的分区数由切片的数量决定的，进而决定了Map Task的数量
     * 当线程数>=2时，默认最小不少于2个分区
     * 当线程数等于1时，则只有一个分区
     */
    val lineRDD: RDD[String] = sc.textFile("spark/data/partitions")

    println(s"lineRDD的分区数为：${lineRDD.getNumPartitions}")

    /**
     * 由于flatMap等其他转换算子不会产生shuffle
     * 则子RDD的分区数默认等于父RDD的分区数
     */
    val wordsRDD: RDD[String] = lineRDD.flatMap(_.split(","))
    println(s"wordsRDD的分区数为：${wordsRDD.getNumPartitions}")

    /**
     * groupBy等一些算子会导致产生shuffle
     * 如果没有手动设置spark.default.parallelism参数，则子RDD的分区数默认也是等于上一个RDD的分区数，但也可以手动指定
     * 手动指定分区数实际上相当于手动指定Reduce Task的数量
     *
     * 优先级：手动指定 > spark.default.parallelism > 上一个RDD的分区数
     */

    // 默认情况
    val grpRDD: RDD[(String, Iterable[String])] = wordsRDD.groupBy(word => word)
    println(s"grpRDD的分区数为：${grpRDD.getNumPartitions}")

    // 手动指定
    val grpRDD2: RDD[(String, Iterable[String])] = wordsRDD.groupBy((word: String) => word, 5)
    println(s"grpRDD的分区数为：${grpRDD2.getNumPartitions}")

    // 手动改变RDD的分区数
    /**
     * repartition vs coalesce
     * repartition 会产生shuffle
     * coalesce 默认不使用shuffle
     *
     * 增加分区必须要有shuffle
     * 减少分区可以有shuffle，也可以没有
     *
     * repartition实际上就是coalesce方法将shuffle参数设为true
     */
    println(s"使用repartition减少grpRDD分区：最终grpRDD分区为：${grpRDD.repartition(2).getNumPartitions}") // 成功了
    println(s"使用repartition增加grpRDD分区：最终grpRDD分区为：${grpRDD.repartition(6).getNumPartitions}") // 成功了

    println(s"使用coalesce减少grpRDD2分区：最终grpRDD2分区为：${grpRDD.coalesce(3).getNumPartitions}") // 成功了
    println(s"使用coalesce增加grpRDD2分区：最终grpRDD2分区为：${grpRDD.coalesce(7).getNumPartitions}") // 失败了


  }

}
